package com.zallds.window;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * 滑动窗口
 * 每间隔2秒钟，统计过去4秒钟内安放在北京西站各个红外测温仪所探测到的体温最高的旅客信息
 */
public class SlideTimeWindowDemo {

    // 第一个time表示步长
    // 第二个time表示间隔
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.socketTextStream("192.168.62.37", 8889)
                .flatMap(new Splitter())
                .keyBy(value -> value.f0)
                .window(SlidingProcessingTimeWindows.of(Time.seconds(4), Time.seconds(2)))
                .maxBy(1)
                .print();
        env.execute();


    }

    public static class Splitter implements FlatMapFunction<String, Tuple2<String, Integer>> {
        @Override
        public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
            String name = s.split(" ")[0];
            Integer age = Integer.parseInt(s.split(" ")[1]);
            collector.collect(new Tuple2<String, Integer>(name, age));
        }
    }
}
